package com.ksyun.campus.client.util;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.ksyun.campus.client.domain.ClusterInfo;
import com.ksyun.campus.client.domain.DataServerMsg;
import com.ksyun.campus.client.domain.MetaServerMsg;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class ZkUtil {
    private static final Logger logger = LoggerFactory.getLogger(ZkUtil.class);
    private static final ObjectMapper objectMapper = new ObjectMapper();
    
    // ZooKeeper连接字符串，默认本地
    private static final String DEFAULT_ZK_CONNECT_STRING = "localhost:2181";
    
    // ZooKeeper路径常量
    private static final String META_SERVER_PATH = "/minfs/meta";
    private static final String META_INSTANCES_PATH = "/minfs/meta/instances";
    private static final String META_LEADER_PATH = "/minfs/meta/leader";
    private static final String META_LEADER_ELECTION_PATH = "/minfs/meta/leader-election";
    private static final String DATA_SERVER_PATH = "/minfs/dataservers";
    private static final String MASTER_PATH = "/minfs/meta";
    
    private ZooKeeper zooKeeper;
    private String connectString;
    
    public ZkUtil() {
        this(DEFAULT_ZK_CONNECT_STRING);
    }
    
    public ZkUtil(String connectString) {
        this.connectString = connectString;
    }
    
    /**
     * 初始化ZooKeeper连接
     */
    public void init() throws IOException, InterruptedException {
        CountDownLatch connectedSignal = new CountDownLatch(1);
        
        zooKeeper = new ZooKeeper(connectString, 5000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    connectedSignal.countDown();
                }
            }
        });
        
        // 等待连接，设置10秒超时
        boolean connected = connectedSignal.await(10, java.util.concurrent.TimeUnit.SECONDS);
        if (!connected) {
            throw new IOException("ZooKeeper连接超时: " + connectString);
        }
        logger.info("ZooKeeper连接成功: {}", connectString);
    }
    
    /**
     * 获取集群信息
     */
    public ClusterInfo getClusterInfo() {
        try {
            ClusterInfo clusterInfo = new ClusterInfo();
            
            // 获取master metaServer信息
            MetaServerMsg masterMetaServer = getMasterMetaServer();
            clusterInfo.setMasterMetaServer(masterMetaServer);
            
            // 获取所有MetaServer实例信息
            List<MetaServerMsg> metaServers = getAllMetaServers();
            clusterInfo.setMetaServers(metaServers);
            
            // 为了兼容性，设置第一个非master实例为slave
            if (metaServers != null && !metaServers.isEmpty()) {
                for (MetaServerMsg metaServer : metaServers) {
                    if (masterMetaServer == null || 
                        !metaServer.getHost().equals(masterMetaServer.getHost()) || 
                        metaServer.getPort() != masterMetaServer.getPort()) {
                        clusterInfo.setSlaveMetaServer(metaServer);
                        break;
                    }
                }
            }
            
            // 获取所有dataServer信息，过滤掉无效的DataServer
            List<DataServerMsg> dataServers = getAllDataServers();
            List<DataServerMsg> validDataServers = new ArrayList<>();
            if (dataServers != null) {
                for (DataServerMsg dataServer : dataServers) {
                    // 过滤掉不存在的DataServer (198.18.0.1:9001)
                    if (!"198.18.0.1".equals(dataServer.getHost()) || dataServer.getPort() != 9001) {
                        validDataServers.add(dataServer);
                    }
                }
            }
            clusterInfo.setDataServer(validDataServers);
            
            return clusterInfo;
        } catch (Exception e) {
            logger.error("获取集群信息失败", e);
            return null;
        }
    }
    
    /**
     * 获取master metaServer信息
     */
    private MetaServerMsg getMasterMetaServer() {
        try {
            logger.debug("检查master节点: {}", MASTER_PATH);
            Stat stat = zooKeeper.exists(MASTER_PATH, false);
            if (stat != null) {
                logger.debug("Master节点存在，获取数据");
                byte[] data = zooKeeper.getData(MASTER_PATH, false, stat);
                if (data != null) {
                    String dataStr = new String(data);
                    logger.debug("Master节点数据: {}", dataStr);
                    if (dataStr.trim().isEmpty()) {
                        logger.warn("Master节点数据为空");
                        return null;
                    }
                    return parseMetaServerData(dataStr);
                } else {
                    logger.warn("Master节点数据为null");
                }
            } else {
                logger.warn("Master节点不存在: {}", MASTER_PATH);
            }
        } catch (Exception e) {
            logger.error("获取master metaServer信息失败", e);
        }
        return null;
    }
    
    /**
     * 获取所有MetaServer实例信息
     */
    private List<MetaServerMsg> getAllMetaServers() {
        List<MetaServerMsg> metaServers = new ArrayList<>();
        try {
            // 先尝试从instances路径获取MetaServer实例
            if (zooKeeper.exists(META_INSTANCES_PATH, false) != null) {
                logger.debug("获取MetaServer实例: {}", META_INSTANCES_PATH);
                List<String> instances = zooKeeper.getChildren(META_INSTANCES_PATH, false);
                logger.debug("找到MetaServer实例: {}", instances);
                
                // 获取所有实例
                for (String instanceId : instances) {
                    String instancePath = META_INSTANCES_PATH + "/" + instanceId;
                    byte[] data = zooKeeper.getData(instancePath, false, null);
                    if (data != null) {
                        String dataStr = new String(data);
                        logger.debug("MetaServer实例 {} 数据: {}", instanceId, dataStr);
                        
                        try {
                            MetaServerMsg metaServer = parseMetaServerInstanceData(dataStr);
                            if (metaServer != null) {
                                // 设置实例名称
                                metaServer.setInstanceId(instanceId);
                                logger.debug("解析到MetaServer实例: {}", metaServer);
                                metaServers.add(metaServer);
                            }
                        } catch (Exception e) {
                            logger.warn("解析MetaServer实例数据失败: {}", e.getMessage());
                        }
                    }
                }
                
                return metaServers;
            }
            
            // 如果instances路径不存在或没有找到实例，回退到原来的逻辑
            logger.debug("获取MetaServer子节点: {}", META_SERVER_PATH);
            List<String> children = zooKeeper.getChildren(META_SERVER_PATH, false);
            logger.debug("找到MetaServer子节点: {}", children);
            
            for (String child : children) {
                String path = META_SERVER_PATH + "/" + child;
                byte[] data = zooKeeper.getData(path, false, null);
                if (data != null) {
                    String dataStr = new String(data);
                    logger.debug("MetaServer {} 数据: {}", child, dataStr);
                    
                    if (dataStr.trim().isEmpty()) {
                        logger.warn("MetaServer {} 数据为空，跳过", child);
                        continue;
                    }
                    
                    MetaServerMsg metaServer = parseMetaServerData(dataStr);
                    if (metaServer != null) {
                        metaServer.setInstanceId(child);
                        metaServers.add(metaServer);
                        logger.debug("找到MetaServer: {}", metaServer);
                    }
                } else {
                    logger.warn("MetaServer {} 数据为null", child);
                }
            }
            
        } catch (Exception e) {
            logger.error("获取MetaServer信息失败", e);
        }
        return metaServers;
    }
    
    /**
     * 解析MetaServer数据，支持IP字符串和JSON格式
     */
    private MetaServerMsg parseMetaServerData(String dataStr) {
        try {
            // 首先尝试解析为JSON
            if (dataStr.trim().startsWith("{")) {
                return objectMapper.readValue(dataStr, MetaServerMsg.class);
            } else {
                // 如果是IP字符串格式，创建MetaServerMsg对象
                MetaServerMsg metaServer = new MetaServerMsg();
                metaServer.setHost(dataStr.trim());
                metaServer.setPort(8300); // 默认端口
                return metaServer;
            }
        } catch (Exception e) {
            logger.error("解析MetaServer数据失败: {}", dataStr, e);
            return null;
        }
    }
    
    /**
     * 解析MetaServer实例数据，从instances节点的JSON格式
     */
    private MetaServerMsg parseMetaServerInstanceData(String dataStr) {
        try {
            if (dataStr.trim().startsWith("{")) {
                // 解析JSON格式的实例数据
                // JSON格式类似: {"instanceId":"metaserver-leader","host":"198.18.0.1","port":8300,"startTime":...,"leader":true}
                com.fasterxml.jackson.databind.JsonNode jsonNode = objectMapper.readTree(dataStr);
                
                MetaServerMsg metaServer = new MetaServerMsg();
                metaServer.setHost(jsonNode.get("host").asText());
                metaServer.setPort(jsonNode.get("port").asInt());
                
                // 可以设置其他属性
                if (jsonNode.has("instanceId")) {
                    // 暂时没有instanceId字段，可以忽略或扩展MetaServerMsg类
                }
                
                return metaServer;
            } else {
                logger.warn("MetaServer实例数据不是JSON格式: {}", dataStr);
                return null;
            }
        } catch (Exception e) {
            logger.error("解析MetaServer实例数据失败: {}", dataStr, e);
            return null;
        }
    }
    
    /**
     * 判断是否为master metaServer
     */
    private boolean isMasterMetaServer(MetaServerMsg metaServer) {
        try {
            Stat stat = zooKeeper.exists(MASTER_PATH, false);
            if (stat != null) {
                byte[] data = zooKeeper.getData(MASTER_PATH, false, stat);
                if (data != null) {
                    String json = new String(data);
                    MetaServerMsg master = objectMapper.readValue(json, MetaServerMsg.class);
                    return master.getHost().equals(metaServer.getHost()) && 
                           master.getPort() == metaServer.getPort();
                }
            }
        } catch (Exception e) {
            logger.error("判断master metaServer失败", e);
        }
        return false;
    }
    
    /**
     * 获取所有dataServer信息
     */
    private List<DataServerMsg> getAllDataServers() {
        List<DataServerMsg> dataServers = new ArrayList<>();
        try {
            List<String> children = zooKeeper.getChildren(DATA_SERVER_PATH, false);
            for (String child : children) {
                String path = DATA_SERVER_PATH + "/" + child;
                byte[] data = zooKeeper.getData(path, false, null);
                if (data != null) {
                    String json = new String(data);
                    logger.debug("DataServer {} JSON数据: {}", child, json);
                    
                    // 检查数据是否为空
                    if (json.trim().isEmpty()) {
                        logger.warn("DataServer {} 的数据为空，跳过", child);
                        continue;
                    }
                    
                    try {
                        DataServerMsg dataServer = objectMapper.readValue(json, DataServerMsg.class);
                        logger.debug("解析到DataServer: {}", dataServer);
                        dataServers.add(dataServer);
                    } catch (Exception e) {
                        logger.error("解析DataServer JSON失败，path: {}, json: {}", path, json, e);
                    }
                }
            }
        } catch (Exception e) {
            logger.error("获取dataServer信息失败", e);
        }
        return dataServers;
    }
    
    /**
     * 监听集群变化
     */
    public void watchClusterChanges(ClusterChangeListener listener) {
        try {
            // 监听master变化
            watchMasterChanges(listener);
            
            // 监听metaServer变化
            watchMetaServerChanges(listener);
            
            // 监听dataServer变化
            watchDataServerChanges(listener);
            
        } catch (Exception e) {
            logger.error("设置集群变化监听失败", e);
        }
    }
    
    private void watchMasterChanges(ClusterChangeListener listener) {
        try {
            zooKeeper.exists(MASTER_PATH, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getType() == Event.EventType.NodeDataChanged || 
                        event.getType() == Event.EventType.NodeDeleted ||
                        event.getType() == Event.EventType.NodeCreated) {
                        listener.onClusterChanged(getClusterInfo());
                        // 重新设置监听
                        watchMasterChanges(listener);
                    }
                }
            });
        } catch (Exception e) {
            logger.error("监听master变化失败", e);
        }
    }
    
    private void watchMetaServerChanges(ClusterChangeListener listener) {
        try {
            zooKeeper.getChildren(META_SERVER_PATH, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getType() == Event.EventType.NodeChildrenChanged) {
                        listener.onClusterChanged(getClusterInfo());
                        // 重新设置监听
                        watchMetaServerChanges(listener);
                    }
                }
            });
        } catch (Exception e) {
            logger.error("监听metaServer变化失败", e);
        }
    }
    
    private void watchDataServerChanges(ClusterChangeListener listener) {
        try {
            zooKeeper.getChildren(DATA_SERVER_PATH, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getType() == Event.EventType.NodeChildrenChanged) {
                        listener.onClusterChanged(getClusterInfo());
                        // 重新设置监听
                        watchDataServerChanges(listener);
                    }
                }
            });
        } catch (Exception e) {
            logger.error("监听dataServer变化失败", e);
        }
    }
    
    /**
     * 关闭ZooKeeper连接
     */
    public void close() {
        if (zooKeeper != null) {
            try {
                zooKeeper.close();
                logger.info("ZooKeeper连接已关闭");
            } catch (InterruptedException e) {
                logger.error("关闭ZooKeeper连接失败", e);
            }
        }
    }
    
    /**
     * 集群变化监听器接口
     */
    public interface ClusterChangeListener {
        void onClusterChanged(ClusterInfo clusterInfo);
    }
}
